package org.rency.mq.kafka.consumer.factory;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.rency.mq.kafka.consumer.handler.AbstractHandler;
import org.rency.mq.kafka.consumer.handler.notify.AbstractBatchNotifyHandler;
import org.rency.mq.kafka.consumer.handler.notify.AbstractNotifyHandler;
import org.rency.mq.kafka.consumer.support.factory.DefaultConsumerFactory;
import org.rency.mq.kafka.consumer.support.factory.KafkaContainerProperties;
import org.rency.mq.kafka.consumer.support.factory.KafkaMessageListenerContainerFactory;
import org.rency.mq.kafka.consumer.support.listener.KafkaBatchMessageListener;
import org.rency.mq.kafka.consumer.support.listener.KafkaMessageListener;
import org.rency.mq.kafka.core.KafkaConstant;
import org.rency.mq.kafka.core.SerializerKind;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.KafkaDataListener;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * ${todo}
 *
 * @author: Created by user_rcy@163.com on 2016/10/21.
 * @version: V1.0.0
 * @since JDK 1.8
 */
public class MultipleMessageHandlerContainer {

    private static final Logger logger = LoggerFactory.getLogger(MultipleMessageHandlerContainer.class);

    @Resource(name="consumerProperties")
    private Map<String, Object>                consumerProperties;

    private List<AbstractMessageListenerContainer<?,?>>    messageListenerContainers = new ArrayList<AbstractMessageListenerContainer<?,?>>();

    @Deprecated
    private List<AbstractBatchNotifyHandler>   batchNotifyHandlerContainer = new ArrayList<AbstractBatchNotifyHandler>();

    private List<AbstractNotifyHandler>        notifyHandlerContainer = new ArrayList<AbstractNotifyHandler>();

    private boolean                            available = true;

    @Resource(name="poolTimeout")
    private long                               pollTimeout;

    /**
     * 启动Kafka监听器
     */
    private void start(){
        logger.info("Kafka Consumer Listener Ready Start.");
        if(!this.available){
            logger.info("Kafka Consumer Listener Switch:[{}].",this.available ? "ON":"OFF");
            return;
        }
        this.init();
    }

    private void stop(){
        for(AbstractMessageListenerContainer<?,?> container : this.messageListenerContainers){
            container.stop();
        }
    }

    private void init(){
        logger.info("Kafka Consumer Listener Init.");
        try {
            for (AbstractHandler batchNotifyHandler : this.batchNotifyHandlerContainer) {
                logger.info("Kafka Consumer Message Container Add Message Handler:[{}]-[{}].",batchNotifyHandler.getHandlerName(),batchNotifyHandler.getClass().getName());
                ConsumerFactory<?, ?> consumerFactory = this.getConsumerFactory(this.consumerProperties,batchNotifyHandler);
                KafkaBatchMessageListener messageListener = (KafkaBatchMessageListener) batchNotifyHandler.getMessageListener();
                messageListener.setMessageHandler(batchNotifyHandler);
                ContainerProperties containerProperties = this.getContainerProperties(batchNotifyHandler,messageListener);
                this.messageListenerContainers.add(new KafkaMessageListenerContainerFactory<>(consumerFactory, containerProperties).getContainer());
            }

            for (AbstractHandler notifyHandler : this.notifyHandlerContainer) {
                logger.info("Kafka Consumer Message Container Add Message Handler:[{}]-[{}].",notifyHandler.getHandlerName(),notifyHandler.getClass().getName());
                ConsumerFactory<?, ?> consumerFactory = this.getConsumerFactory(this.consumerProperties,notifyHandler);
                KafkaMessageListener messageListener = (KafkaMessageListener) notifyHandler.getMessageListener();
                messageListener.setMessageHandler(notifyHandler);
                ContainerProperties containerProperties = this.getContainerProperties(notifyHandler, messageListener);
                this.messageListenerContainers.add(new KafkaMessageListenerContainerFactory<>(consumerFactory, containerProperties).getContainer());
            }

            for (AbstractMessageListenerContainer<?, ?> container : this.messageListenerContainers) {
                container.start();
            }
        }catch (IllegalArgumentException e){
            logger.error("Kafka Consumer Property Validation Failed.",e);
            throw new RuntimeException("Kafka Consumer Validation Failed With Start!",e);
        }catch (Exception e){
            logger.error("Kafka Consumer Start Failed.",e);
            throw new RuntimeException("Kafka Consumer Start Failed.",e);
        }
    }

    private ConsumerFactory<?, ?> getConsumerFactory(Map<String, Object> properties,AbstractHandler notifyHandler){
        Assert.hasText(notifyHandler.getGroupName(),"Kafka Consumer Group Name Must Be Provided!");
        Assert.notNull(notifyHandler.getKeyGenericClass(),"Kafka Consumer Listen Key Class type Must Be Provided!");
        Assert.notNull(notifyHandler.getValueGenericClass(),"Kafka Consumer Listen Data Class type Must Be Provided!");
        Map<String, Object> param = new HashMap<>();
        param.putAll(properties);
        param.put(ConsumerConfig.GROUP_ID_CONFIG, notifyHandler.getGroupName());
        param.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, SerializerKind.get(notifyHandler.getKeyGenericClass()).getDeserializer());
        param.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, SerializerKind.get(notifyHandler.getValueGenericClass()).getDeserializer());
        return new DefaultConsumerFactory<>(param);
    }

    private ContainerProperties getContainerProperties(AbstractHandler notifyHandler,KafkaDataListener messageListener) throws Exception {
        Assert.hasText(notifyHandler.getTopicName(),"Kafka Consumer Listen Topic Name Must Be Provided!");
        logger.info("Kafka Consumer Listen Topic:[{}].",notifyHandler.getTopicName());
        Assert.notNull(notifyHandler.getMessageListener(),"Kafka Consumer Listener Must Be Provided!");
        ContainerProperties containerProperties = new KafkaContainerProperties(notifyHandler.getTopicName(), notifyHandler.getTaskExecutor(), messageListener).getObject();
        containerProperties.setAckMode(notifyHandler.getAckMode());
        if (KafkaConstant.DEFAULT_CONSUMER_POLL_TIMEOUT < this.pollTimeout) {
            containerProperties.setPollTimeout(this.pollTimeout);
        }
        return containerProperties;
    }

    @Deprecated
    public void setBatchNotifyHandlerContainer(List<AbstractBatchNotifyHandler> batchNotifyHandlerContainer) {
        this.batchNotifyHandlerContainer = batchNotifyHandlerContainer;
    }

    public void setNotifyHandlerContainer(List<AbstractNotifyHandler> notifyHandlerContainer) {
        this.notifyHandlerContainer = notifyHandlerContainer;
    }

    public void setAvailable(boolean available) {
        this.available = available;
    }
}
